package edu.zju.gis.dbfg.model.loader;

import edu.zju.gis.dbfg.model.util.*;
//import edu.zju.gis.dbfg.model.util.*;
import org.apache.spark.broadcast.Broadcast;
import org.geotools.geometry.jts.JTS;
import org.geotools.referencing.CRS;
import org.opengis.referencing.crs.CoordinateReferenceSystem;
import org.opengis.referencing.operation.MathTransform;
import com.google.gson.Gson;
import com.google.gson.GsonBuilder;
import org.json.JSONArray;
import org.locationtech.jts.geom.Coordinate;
import org.locationtech.jts.geom.Geometry;
import org.locationtech.jts.geom.GeometryFactory;
import org.locationtech.jts.io.ParseException;
import org.locationtech.jts.io.WKTReader;
//import com.vividsolutions.jts.geom.Coordinate;
//import com.vividsolutions.jts.geom.Geometry;
//import com.vividsolutions.jts.geom.GeometryFactory;
//import com.vividsolutions.jts.io.WKTReader;
import edu.zju.gis.dbfg.model.BaseModel;
import edu.zju.gis.dbfg.model.entity.*;
import edu.zju.gis.dbfg.model.exception.ModelFailedException;
import lombok.Getter;
import lombok.Setter;
import org.apache.spark.SparkConf;
import org.apache.spark.api.java.JavaPairRDD;
import org.apache.spark.api.java.JavaRDD;
import org.apache.spark.api.java.function.PairFlatMapFunction;
import org.apache.spark.util.DoubleAccumulator;
import org.apache.spark.util.LongAccumulator;
import org.geotools.geojson.geom.GeometryJSON;
import org.json.JSONException;
import org.json.JSONObject;
import org.mortbay.util.ajax.JSON;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.Tuple2;

import java.sql.*;

import java.io.File;
import java.io.IOException;
//import java.text.ParseException;
import java.text.SimpleDateFormat;
import java.util.*;
import java.util.Date;
import java.util.stream.Collectors;


/**
 * @author Hu
 * @date 2020/1/4
 * Elasticsearch-spark 的 error handlers 回调，在 es 6.4 版本才有
 * 因此这里只能自己实现各个 partition 的入库
 **/
@Getter
@Setter
public class ESDataLoader extends BaseModel<edu.zju.gis.dbfg.model.loader.ESDataLoaderArgs> {

    private final static Logger logger = LoggerFactory.getLogger(ESDataLoader.class);
    private final static String ERROR_F = "ERROR";
    private final static String TUPLE_F = "TUPLE";
    private final static String RANDOM_KEY_FIELD_PREFIX = "RANDOM_ID_";

    private final static Integer STEP_SIZE = 1000;

    private String esIndex;
    private String type;
    private ConstructedDataset dataset;
    private String uuid;
    private String logs;
//    private int successNumBatch;
    //    private String crs;
//    private Connection con;
//    private Statement statement;

    public ESDataLoader(SparkSessionType type, String appName, String[] args) {
        super(type, appName, args);
    }

    public ESDataLoader(SparkSessionType type, String appName, SparkConf sc, String[] args) {
        super(type, appName, sc, args);
    }

    public static void main(String[] args) throws IOException {
        System.setProperty("es.set.netty.runtime.available.processors", "false");
        ESDataLoader loader = new ESDataLoader(SparkSessionType.SHELL, "ESDataLoader", args);
        loader.run();
    }

    @Override
    protected void initArg(String[] args) {
        super.initArg(args);
//       this.crs=this.arg.getCrs();
        this.esIndex = this.arg.getOutput().split("/")[0];
        this.type = this.arg.getOutput().split("/")[1];
        Gson gson = new GsonBuilder().setDateFormat("yyyy-MM-dd HH:mm:ss").create();
        this.dataset = gson.fromJson(this.arg.getMeta(), ConstructedDataset.class);
        //向数据库插入一条数据
//        Connection con;
//        String driver = "com.mysql.cj.jdbc.Driver";
//        String url = "jdbc:mysql://101.132.255.121:3306/shengting2?allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8";
//        String user = "root";
//        String password = "root";
//        try {
//            //注册JDBC驱动程序
//            Class.forName(driver);
//            //建立连接
//            con = DriverManager.getConnection(url, user, password);
//            if (!con.isClosed()) {
//                System.out.println("数据库连接成功");
//
                this.uuid = this.dataset.getId();
//                SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
////                    System.out.println(df.format(new Date()));
//                JSONArray jsonArray = new JSONArray();
//                for (ConstructedDataField constructedDataField : this.dataset.getFields()) {
//                    JSONObject jsonObject = new JSONObject(constructedDataField);
//                    jsonArray.put(jsonObject);
//                }
////          System.out.println(jsonArray.toString());
//                Statement statement = con.createStatement();
//             String departLevelSql="select depart,usrlevel from tb_role left join tb_user on tb_user.role=tb_role.id where tb_user.id='"+this.arg.getUserId()+"'";
//                ResultSet rs = statement.executeQuery(departLevelSql);
//                String  sql="";
//                while(rs.next()){
//                    System.out.println(rs.getString(1)+rs.getString(2));
//                    sql = "insert into tb_constructed_data VALUES('" + this.uuid + "','"+this.arg.getUserId()+"','" + type + "',-1,'" + this.dataset.getDescription() + "','" + this.dataset.getPicurl() + "','" + jsonArray.toString() + "','" + df.format(new Date()) + "','" + df.format(new Date()) + " ',1,'polygon','ES',999,'" + rs.getString(1)+"')";
//                }
//                String sql2 = "insert into tb_loader_monitor VALUES('" + this.uuid + "','" + type + "','" + df.format(new Date()) + "',0,0,'Loading',' ','"+this.arg.getUserId()+"')";
//                int resultSet = statement.executeUpdate(sql);
//                int resultSet2 = statement.executeUpdate(sql2);
//                if (resultSet == 1 && resultSet2 == 1) {
//                    System.out.println("数据插入成功");
//                }
//
//            }
//            con.close();
//        } catch (ClassNotFoundException e) {
//            System.out.println("数据库驱动没有安装");
//        } catch (SQLException e) {
//            e.printStackTrace();
//            System.out.println("数据库连接失败");
//        }

    }

    public Boolean validate(Map<ConstructedDataField, String> eachRow) {
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getType().equals(ConstructedDataFieldType.Integer.name())) {
                if (!ValidateUtils.isInteger(eachRow.get(field))) {
                    return false;
                }
            } else if (field.getType().equals(ConstructedDataFieldType.Float.name())
                    || field.getType().equals(ConstructedDataFieldType.Long.name())
                    || field.getType().equals(ConstructedDataFieldType.Double.name())) {
                if (!ValidateUtils.isDecimal(eachRow.get(field))) {
                    if (eachRow.get(field).length() == 0) {
                        return true;
                    }
                    return false;
                }
            }
        }
        return true;
    }

    private String getFormatValue(ConstructedDataField field, String value) throws ParseException {
        if (field.getIsTime()) {
            String format = field.getFormat();
            try {
                Long slong = DateUtil.parse(value, format).getTime();
                return Long.toString(slong);
            } catch (Exception e) {

            }
        }
        return value;
    }

    // 对于特殊字段，如datetime以及geometry(lat,lon,start_lat,start_lon)等特殊处理
    public Map<ConstructedDataField, String> mapToConstructedDatafield(String line) throws ParseException {
//        System.out.println("RRRRRRRRRRRRR"+line);
        List<ConstructedDataField> metaFields = this.dataset.getFields();
        // 循环遍历数据，将数据转换为dataset定义的字段中
        String regex = line.contains("\t") ? "\t" : ",";
        Map<ConstructedDataField, String> tuple = new HashMap<>();
        String[] srcFields = line.split(regex);
        for (ConstructedDataField field : metaFields) {
            if (field.getName().startsWith(RANDOM_KEY_FIELD_PREFIX)) {
                tuple.put(field, UUID.randomUUID().toString());
            } else {
                tuple.put(field, getFormatValue(field, srcFields[field.getIndex()]));
            }
        }
        return tuple;
    }

    /**
     * @param value
     * @return _1: id; _2: value in json format
     */
    private Tuple2<String, String> mapToJson(Map<ConstructedDataField, String> value, MathTransform transform) throws Exception {
        String key = null;
        JSONObject json = new JSONObject();
        for (Map.Entry<ConstructedDataField, String> v : value.entrySet()) {
            ConstructedDataField field = v.getKey();
            if (field.getIsKey()) {
                key = v.getValue();
            }
            if (field.getType().equals("String") || field.getType().equals(DataType.STRING.name())) {
                json.put(field.getName(), v.getValue());
                //空间字段必须叫GEOM或者GEOM_ST或GEOM_ED
                if (field.getName().equals(FeatureType.SPATIAL_FIELD_NAME) || field.getName().equals(FeatureType.SPATIAL_OD_ED_NAME) || field.getName().equals(FeatureType.SPATIAL_OD_ST_NAME)) {
                    String wkt = this.getWktGeometry(dataset.getDatatype(), field.getName(), value);
                    try {
                        WKTReader reader = new WKTReader();
                        Geometry geometry = reader.read(wkt);
                        if (transform != null) {
                            geometry = JTS.transform(geometry, transform);
                        }
                        GeometryJSON g = new GeometryJSON(9);
                        String geojson = g.toString(geometry);
                        JSONObject geo = new JSONObject(geojson);
                        json.put(field.getName(), geo);


//                        WKTReader reader = new WKTReader();
//                        Geometry geometry = reader.read(wkt);
////
//                        Geometry geometry2 = JTS.transform(geometry, transform);
//                        GeometryJSON g = new GeometryJSON(9);
//                        String geojson = g.toString(geometry2);
////                        System.out.println("！！！geojson的值是"+geojson);
//                        JSONObject geo = new JSONObject(geojson);
//                        json.put(field.getName(), geo);
                    } catch (Exception e) {
                        System.out.println("************************");
                        System.out.println(e.getMessage());
                        System.out.println("************************");
                    }


                }
            } else if (field.getType().equals("Integer") || field.getType().equals(DataType.INT.name())) {
                json.put(field.getName(), Integer.valueOf(value.get(field)));
            } else if (field.getType().equals("Float") || field.getType().equals(DataType.FLOAT.name())) {
                json.put(field.getName(), Float.valueOf(value.get(field)));
            } else if (field.getType().equals("Double") || field.getType().equals(DataType.DOUBLE.name())) {
                if (value.get(field).length() == 0) {
                    json.put(field.getName(), 0.00);
                } else {
                    json.put(field.getName(), Double.valueOf(value.get(field)));
                }
            } else if (field.getType().equals("Long") || field.getType().equals(DataType.LONG.name())) {
                json.put(field.getName(), Long.valueOf(value.get(field)));
            } else if (field.getType().equals("Boolean") || field.getType().equals(DataType.BOOLEAN.name())) {
                json.put(field.getName(), Boolean.valueOf(value.get(field)));
            } else {
                throw new UnsupportedOperationException("Unvalid field type: " + field.getType());
            }
        }

        if (key == null) {
            throw new UnsupportedOperationException("must have key field: " + json.toString());
        }

        return new Tuple2<>(key, json.toString());
    }

    private String getWktGeometry(String spatialType, String fieldName, Map<ConstructedDataField, String> eachRow) {
        switch (ConstructedDataType.valueOf(spatialType)) {
            case POINT:
                return getPointWKT(eachRow);
            case POLYGON:
                return getPolygonWKT(eachRow);
            case POLYLINE:
                return getPolyLineWKT(eachRow);
            case ODPAIR:
                return getOdPairWKT(fieldName, eachRow);
        }
        throw new UnsupportedOperationException("unsupport geometry type: " + spatialType);
    }

    //点数据支持3种格式数据的标准化
    //单字段wkt
    //双字段lon,lat
    //逗号隔开的单字段lon,lat
    private String getPointWKT(Map<ConstructedDataField, String> eachRow) {
        ConstructedDataField ptLon = null;
        ConstructedDataField ptLat = null;
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getFormat().equals(FeatureType.WKT_FIELD)) {
                return eachRow.get(field);
            }
            if (field.getFormat().equals(FeatureType.POINT_FIELD_CORS)) {
                try {
                    String[] lonlat = eachRow.get(field).split(",");
                    return lonlat2wktPoint(lonlat[0], lonlat[1]);
                } catch (Exception e) {
                    throw new UnsupportedOperationException("Unvalid cors:" + eachRow.get(field));
                }
            }
            if (field.getFormat().equals(FeatureType.POINT_FIELD_LON))
                ptLon = field;
            if (field.getFormat().equals(FeatureType.POINT_FIELD_LAT))
                ptLat = field;
        }
        if (ptLat != null & ptLon != null) {
            return lonlat2wktPoint(eachRow.get(ptLon), eachRow.get(ptLat));
        }
        throw new UnsupportedOperationException("Can not get the Point By: WKT,Lon/Lat,Cors");
    }

    //面数据支持2种格式数据的标准化
    //单字段wkt
    //单字段cors: lon1 lat1,lon2 lat2,lon3 lat3 ? 无法解析内环。
    private String getPolygonWKT(Map<ConstructedDataField, String> eachRow) {
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getFormat().equals(FeatureType.WKT_FIELD)) {
                return eachRow.get(field);
            }
            if (field.getFormat().equals(FeatureType.POLYGON_FIELD)) {
                return corsStr2wkt(FeatureType.POLYGON, eachRow.get(field));
            }
        }
        throw new UnsupportedOperationException("Can not get the Polygon By: WKT,Cors");
    }




    //线数据支持2种格式数据的标准化
    //单字段wkt
    //单字段cors: lon1 lat1,lon2 lat2,lon3 lat3
    private String getPolyLineWKT(Map<ConstructedDataField, String> eachRow) {
        for (ConstructedDataField field : eachRow.keySet()) {
            if (field.getFormat().equals(FeatureType.WKT_FIELD)) {
                return eachRow.get(field);
            }
            if (field.getFormat().equals(FeatureType.LINESTRING_FIELD)) {
                return corsStr2wkt(FeatureType.LINESTRING, eachRow.get(field));
            }
        }
        throw new UnsupportedOperationException("Can not get the Polyline By: WKT,Cors");
    }

    //OD数据支持1种格式数据的标准化
    //四字段startLon,startLat,endLon,endLat
    private String getOdPairWKT(String fieldName, Map<ConstructedDataField, String> eachRow) {
        ConstructedDataField stLon = null;
        ConstructedDataField stLat = null;
        ConstructedDataField edLon = null;
        ConstructedDataField edLat = null;
        for (ConstructedDataField field : eachRow.keySet()) {
            switch (field.getFormat()) {
                case FeatureType.ODPAIR_FIELD_SLON:
                    stLon = field;
                    break;
                case FeatureType.ODPAIR_FIELD_SLAT:
                    stLat = field;
                    break;
                case FeatureType.ODPAIR_FIELD_ELON:
                    edLon = field;
                    break;
                case FeatureType.ODPAIR_FIELD_ELAT:
                    edLat = field;
                default:
                    break;
            }
        }
        switch (fieldName) {
            case FeatureType.SPATIAL_FIELD_NAME:
                if (stLon != null & stLat != null & edLon != null & edLat != null) {
                    return od2wktPolyLine(eachRow.get(stLon), eachRow.get(stLat), eachRow.get(edLon), eachRow.get(edLat));
                } else {
                    return "";
                }
            case FeatureType.SPATIAL_OD_ST_NAME:
                if (stLon != null & stLat != null) {
                    return lonlat2wktPoint(eachRow.get(stLon), eachRow.get(stLat));
                } else {
                    return "";
                }
            case FeatureType.SPATIAL_OD_ED_NAME:
                if (edLon != null & edLat != null) {
                    return lonlat2wktPoint(eachRow.get(edLon), eachRow.get(edLat));
                } else {
                    return "";
                }
            default:
                return "";
        }
    }

    private String lonlat2wktPoint(String lon, String lat) {
        try {
            return new GeometryFactory().createPoint(new Coordinate(Double.valueOf(lon), Double.valueOf(lat))).toText();
        } catch (Exception e) {
            throw new UnsupportedOperationException("Unvalid cors: " + lon + "," + lat);
        }
    }

    private String corsStr2wkt(String type, String cors) {
        try {
            String[] corsArray = cors.split(",");
            Coordinate[] cords = new Coordinate[corsArray.length];
            for (int i = 0; i < corsArray.length; i++) {
                String[] lonlat = corsArray[i].split(" ");
                cords[i] = new Coordinate(Double.valueOf(lonlat[0]), Double.valueOf(lonlat[1]));
            }
            if (FeatureType.POLYGON.equals(type))
                return new GeometryFactory().createPolygon(cords).toText();
            if (FeatureType.LINESTRING.equals(type))
                return new GeometryFactory().createLineString(cords).toText();
            throw new UnsupportedOperationException("Unsupport geometry type for corsStr2wkt(): " + type);
        } catch (Exception e) {
            throw new UnsupportedOperationException("Unvalid cors: " + cors);
        }
    }

    private String od2wktPolyLine(String stLon, String stLat, String edLon, String edLat) {
        try {
            Coordinate[] cords = new Coordinate[2];
            cords[0] = new Coordinate(Double.valueOf(stLon), Double.valueOf(stLat));
            cords[1] = new Coordinate(Double.valueOf(edLon), Double.valueOf(edLat));
            return new GeometryFactory().createLineString(cords).toText();
        } catch (Exception e) {
            throw new UnsupportedOperationException("Unvalid od: " + stLon + stLat + edLat + edLat);
        }
    }

    // 基于 ConstructedDataField，构建入库数据json
    private Tuple2<String, Object> mapValues(String line) {

        if (line == null || line.trim().length() == 0) {
            System.out.println("fffffffff" + line);
            return new Tuple2<>(ERROR_F, "empty line");
        }

        Map<ConstructedDataField, String> tuple = new HashMap<>();
        try {
            tuple = mapToConstructedDatafield(line);
            if (!validate(tuple)) {
                return new Tuple2<>(ERROR_F, "validate failed: " + line);
            }
        } catch (Exception e) {
            return new Tuple2<>(ERROR_F, "\nUnKnown Error:" + e.getMessage());
        }

        return new Tuple2<>(TUPLE_F, tuple);
    }

    /**
     * 基于 ConstructdDataset 构建 es mapping
     *
     * @return
     */
    private String datasetToEsMapping() {
        List<ConstructedDataField> fields = this.dataset.getFields();
        Map<String, Object> properties = new HashMap<>();
        for (ConstructedDataField field : fields) {
            Map<String, String> fieldMap = new HashMap<>();
            if (field.getType().equals("String") || field.getType().equals(DataType.STRING.name())) {
                if (field.getIsGeometry() ||  field.getName().equals(FeatureType.SPATIAL_FIELD_NAME)) {
                    fieldMap.put("type", "geo_shape");
                } else if (field.getName().equals(FeatureType.SPATIAL_OD_ST_NAME) || field.getName().equals(FeatureType.SPATIAL_OD_ED_NAME)) {
                    fieldMap.put("type", "geo_shape");
                } else if (field.getIsTime()) {
                    fieldMap.put("type", "date");
                    fieldMap.put("format", "epoch_millis");
                } else {
                    fieldMap.put("type", "text");
                }
                if (field.getIsChinese()) {
                    fieldMap.put("analyzer", "ik_max_word");
                    fieldMap.put("search_analyzer", "ik_smart");
                }
            } else if (field.getType().equals("Integer") || field.getType().equals(DataType.INT.name())) {
                fieldMap.put("type", "integer");
            } else if (field.getType().equals("Float") || field.getType().equals(DataType.FLOAT.name())) {
                fieldMap.put("type", "float");
            } else if (field.getType().equals("Double") || field.getType().equals(DataType.DOUBLE.name())) {
                fieldMap.put("type", "double");
            } else if (field.getType().equals("Boolean") || field.getType().equals(DataType.BOOLEAN.name())) {
                fieldMap.put("type", "boolean");
            } else if (field.getType().equals("Long") || field.getType().equals(DataType.LONG.name())) {
                fieldMap.put("type", "long");
            } else {
                throw new ModelFailedException(ESDataLoader.class, "datasetToEsMapping()", "Unsupport field type", field.getType());
            }
            properties.put(field.getName(), fieldMap);
        }
        Map<String, Object> mapping = new HashMap<>();
        mapping.put("properties", properties);
        Gson gson = new Gson();
        return gson.toJson(mapping);
    }

    private List<Tuple2<String, String>> batchLoad(List<Map<ConstructedDataField, String>> ts, int count, MathTransform transform) throws IOException {

        List<Tuple2<String, String>> result = new ArrayList<>();
        Gson gson = new Gson();

        if (count >= EsHelper.RETRY_TIME) {
            for (Map<ConstructedDataField, String> t : ts) {
                result.add(new Tuple2<>(ERROR_F, "Retry time: " + count + ", Abort: " + gson.toJson(t)));

            }
            return result;
        } else if (count > 1) {
            for (Map<ConstructedDataField, String> t : ts) {

                logger.info("Retry time:" + count + ": " + gson.toJson(t));

            }
        }

        EsHelper helper = EsHelper.getInstanceClient();

        List<Tuple2<String, String>> tss = new ArrayList<>();
//        System.out.println("ts的size" + ts.size());
        for (Map<ConstructedDataField, String> t : ts) {
            try {
//                System.out.println("t的值是"+t.toString());
                tss.add(mapToJson(t, transform));
            } catch (Exception e) {
//                System.out.println("!!!!!!!!!!!!!!异常是"+e.getMessage());
                result.add(new Tuple2<>(ERROR_F, "map to json failed: " + e.getMessage()));
            }
        }
//        System.out.println("tss的size" + tss.size());
//        System.out.println("ESINDEX:TYPE:TSS" + esIndex + type + tss);
        Map<String, String> errors = helper.batchLoad(esIndex, type, tss);

        // retry error objects
        // TODO 由于类型不一致，导致多 O(n) 次 geojson 序列化
        if (errors.size() != 0) {
            count = count + 1;
            List<Map<ConstructedDataField, String>> errorTs = new ArrayList<>();
            for (Map<ConstructedDataField, String> t : ts) {
                String key = t.get(this.dataset.getKeyField());
                if (errors.containsKey(key)) {
                    errorTs.add(t);
                }
            }
            result.addAll(batchLoad(errorTs, count, transform));
        }

//        if(con!=null){
//
//        }


        return result;
    }

    private boolean createIndex(EsHelper helper, String mapping) {
        return helper.createIndex(this.esIndex, this.type, mapping);
    }

    @Override
    public void run() throws IOException {
        this.logs = "";
        EsHelper helper = EsHelper.getInstanceClient();
        try {
            ConstructedDataField keyField = this.dataset.getKeyField();
            if (keyField == null) {
                keyField = new ConstructedDataField();
                keyField.setIsKey(true);
                keyField.setName(RANDOM_KEY_FIELD_PREFIX + UUID.randomUUID().toString());
                keyField.setType("String");
                keyField.setText("PK");
            }
            this.dataset.getFields().add(keyField);
            System.out.println(this.dataset.getFields().size());
            // 构建索引
            this.createIndex(helper, datasetToEsMapping());
            // 停止索引更新
            helper.stopRefresh(this.esIndex);
            System.out.println("数据开始入库");
            // 数据转化为json

            JavaPairRDD<String, Object> inDataRDD = this.jsc.textFile(this.arg.getInput()).mapToPair(this::mapValues);

//            System.out.println("FFFFFFFF"+inDataRDD.collectAsMap().toString());
            JavaRDD<String> errorsRDD = inDataRDD.filter(in -> in._1.equals(ERROR_F)).map(in -> (String) in._2);

            JavaRDD<Object> jsonRDD = inDataRDD.filter(in -> !in._1.equals(ERROR_F)).map(in -> in._2);

            LongAccumulator total = this.jsc.sc().longAccumulator("total_count");

//            CoordinateReferenceSystem crs1 = CRS.parseWKT(CrsPick.pickByName(this.arg.getCrs()));
//            Broadcast<MathTransform> tstransformBroad = jsc.broadcast(transform1);

            // 转化为 json 并导入
            JavaPairRDD<String, String> loadRDD = jsonRDD.mapPartitionsToPair(new PairFlatMapFunction<Iterator<Object>, String, String>() {
                @Override
                public Iterator<Tuple2<String, String>> call(Iterator<Object> in) throws Exception {

                    Connection con;
                    String driver = "com.mysql.cj.jdbc.Driver";
//                    String url = "jdbc:mysql://101.132.255.121:3306/shengting2?allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8";
//                    String user = "root";
//                    String password = "root";
                    String url = "jdbc:mysql://192.168.1.5:3306/shengting2?allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8";
                    String user = "root";
                    String password = "ubt123";
                    List<Map<ConstructedDataField, String>> batch = new ArrayList<>();
                    List<Tuple2<String, String>> result = new ArrayList<>();
                    int count = 0;
                    try {
                        Class.forName(driver);
                        //建立连接
                        con = DriverManager.getConnection(url, user, password);
                        Statement statement = con.createStatement();
                        if (!con.isClosed()) {
//                    String crs = "EPSG:" + ESDataLoader.this.arg.getCrs();
//                    CoordinateReferenceSystem crs1 = CRS.decode(crs, true);
//                    CoordinateReferenceSystem targetCRS = CRS.decode("EPSG:4326", true);
                            String crs = ESDataLoader.this.arg.getCrs();
                            CoordinateReferenceSystem targetCRS = CRS.parseWKT(CrsPick2.pickByName("4326"));
                            String crss = CrsPick2.pickByName(crs);
                            CoordinateReferenceSystem crs1 = CRS.parseWKT(crss);
                            MathTransform transform1 = null;
                            if (!crs1.equals(targetCRS)) {
                                transform1 = CRS.findMathTransform(crs1, targetCRS, true);
                            }
                            System.out.println("进入循环前");
//                        if (!con.isClosed()) {

                            while (in.hasNext()) {
//                        System.out.println("进入循环");
                                count++;
//                        System.out.println("count之后！！！！！！！");
                                Map<ConstructedDataField, String> t = (Map<ConstructedDataField, String>) in.next();
//                        System.out.println(t.values());
                                batch.add(t);
//                        System.out.println(batch.size());
                                // 默认 step-size 为 1000
                                if (batch.size() == STEP_SIZE) {
                                    result.addAll(batchLoad(batch, 1, transform1));
                                    //成功录入的条数是batch-batchLoad
                                   int successNumBatch = batch.size() - batchLoad(batch, 1, transform1).size();
                                    String mysql33 = "UPDATE tb_loader_monitor set successNumb=successNumb+" +successNumBatch + " where id='" + ESDataLoader.this.uuid + "'";
////                                    Statement statement = con.createStatement();
                                    int resultTemp = statement.executeUpdate(mysql33);
//                                    if (resultTemp == 1) {
//                                        System.out.println("数据库更新成功！！！");
//                                    }
                                    batch.clear();
                                }
                            }

//                    System.out.println("batch的值是"+batch.size());
                            result.addAll(batchLoad(batch, 1, transform1));
                            total.add(count);
//                            int successNumBatch2 = batch.size() - batchLoad(batch, 1, transform1).size();
//                            String mysql3 = "UPDATE tb_loader_monitor set successNumb=successNumb+" + successNumBatch2+ " where id='" + ESDataLoader.this.uuid + "'";
//                            int resultTemp2 = statement.executeUpdate(mysql3);
//                            if (resultTemp2 == 1) {
//                                System.out.println("数据库更新成功！！！");
//                            }
                            //成功录入的条数是batch-batchLoad
                            return result.iterator();
                        }
                        con.close();
                    } catch (ClassNotFoundException e) {
                        System.out.println("数据库驱动没有安装");
                    } catch (SQLException e) {
                        e.printStackTrace();
                        System.out.println("数据库连接失败");
                    }
                    return result.iterator();
                }
            });

            List<String> loadError = loadRDD.map(in -> in._2).collect();
            List<String> transformError = errorsRDD.collect();
            total.add(transformError.size());
//            System.out.println("数据入库结束");
            // TODO 与数据库交互
            logger.info("Total count: " + total.value());
            logger.info("Transform Error count: " + transformError.size());
            logger.info("Load Error count: " + loadError.size());

            logger.info("===== TRANSFORM ERROR =====");
            transformError.forEach(x -> logger.info(x));
            if (transformError.size() != 0) {
                this.logs = this.logs + "!transformError!" + "\r";
                for (int i = 0; i < transformError.size(); i++) {
                    this.logs = this.logs + transformError.get(i) + "\r";
                }
            }

            logger.info("===== LOAD ERROR =====");
            loadError.forEach(x -> logger.info(x));
            if (loadError.size() > 0) {
                this.logs = this.logs + "!loadError!" + "\r";
                for (int i = 0; i < loadError.size(); i++) {
                    this.logs = this.logs + loadError.get(i) + "\r";
                }
            }
//            try {
            //注册JDBC驱动程序
            Connection con;
            String driver = "com.mysql.cj.jdbc.Driver";
//            String url = "jdbc:mysql://101.132.255.121:3306/shengting2?allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8";
//            String user = "root";
//            String password = "root";
            String url = "jdbc:mysql://192.168.1.5:3306/shengting2?allowMultiQueries=true&useUnicode=true&characterEncoding=UTF-8&useSSL=false&serverTimezone=GMT%2B8";
            String user = "root";
            String password = "ubt123";
            try {
                Class.forName(driver);
                //建立连接
                con = DriverManager.getConnection(url, user, password);
                Statement statement = con.createStatement();
                if (!con.isClosed()) {
//                    System.out.println("数据库第二次连接成功");
                    Long successNum = total.value() - transformError.size() - loadError.size();
                    int errorNum = transformError.size() + loadError.size();
                    SimpleDateFormat df = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss");//设置日期格式
                    String sql = "UPDATE tb_constructed_data SET size=" + successNum + ",lastModifyTime='" + df.format(new Date()) + "' where id='" + this.uuid + "'";
                    String sql2 = "UPDATE tb_loader_monitor set successNumb=" + successNum + ",errorNumb=" + errorNum + ",`status`='Finish',`logs`='" + this.logs + "' where id='" + this.uuid + "'";
//                    Statement statement = con.createStatement();
                    int resultSet = statement.executeUpdate(sql);
                    int resultSet2 = statement.executeUpdate(sql2);
                    if (resultSet == 1 && resultSet2 == 1) {
                        System.out.println("数据更新成功");
                    }
                }
                con.close();
            } catch (ClassNotFoundException e) {
                System.out.println("数据库驱动没有安装");
            } catch (SQLException e) {
                e.printStackTrace();
                System.out.println("数据库连接失败");
            }
            // 更新索引
            helper.startRefresh(this.esIndex);
        } catch (Exception e) {
//            throw e;
        } finally {
            helper.finish();
            this.close();
        }


    }

    @Override
    protected void close() {
        super.close();
    }

}
